<?php
/**
 * 多进程
 */

namespace Push\Controller;


use Common\Libiary\LogEX;
use Common\Libiary\WechatMaterial;
use Think\Controller;




class MPushController extends Controller
{

    const MAX_PROCESS_NUM = 10;
    const FANS_INDEX_KEY = 'oper:tags:cli:push';
    const FANS_INDEX_FAIL_KEY = 'oper:tags:cli:push:fail';
    private $subProcesses = 0;
    private $redis = null;
    private $needSleep = true;
    private $ppid = 0;

    function __construct(){
        if (substr(php_sapi_name(), 0, 3) !== 'cli') {
            die("This Programe can only be run in CLI mode");
        }
        set_time_limit(0);
        ini_set('seaslog.use_buffer','0');
        ini_set('seaslog.buffer_size','0');
        set_error_handler(array($this,'err_handle'));
        parent::__construct();
        $this->redis = $this->getRedis();
        $this->ppid = posix_getpid();
    }

    function init(){
        while(true){
            if($this->needSleep) sleep(1);
            $this->show("sleep:1  subProcess num:". $this->subProcesses);
            $cont = $this->keepFetchKey();
            if(empty($cont) && $this->subProcesses == 0){
                $this->needSleep = true;
                continue;
            }
            if(empty($cont) && $this->subProcesses > 0){
                $this->show("wait subProcess end  num:". $this->subProcesses);
                pcntl_wait($status);
                $this->subProcesses--;
                continue;
            }
            $this->controlProcess($cont);
            $this->needSleep = false;
        }
    }
    function controlProcess($cont){
        $pid = pcntl_fork();
        if($pid == 0){
            //子进程运行代码
            $this->show("in subProcess");
            $this->controlSubtask($cont);
        }elseif($pid==-1){
            $this->show("create subProcess fail");
            $this->controlProcess($cont);
        }else{
            //父进程运行代码,达到上限时父进程阻塞等待任一子进程退出后while循环继续
            $this->subProcesses++;
            $this->show("subProcess num:".$this->subProcesses);
            if($this->subProcesses>= self::MAX_PROCESS_NUM){
                $this->show("wait subProcess");
                pcntl_wait($status);
                $this->subProcesses--;
            }
            $this->show("in fatProcess");
        }
    }
    function keepFetchKey(){
        try{
            $cont  =$this->redis->rPop(self::FANS_INDEX_KEY);
            if($cont=='+OK'){
                $this->show('get redisCount:'.$cont,'NullRedis');
                return '';
            }
            return $cont;
        }catch(\Exception $e){
            $this->show('reconnect redis err:'.$e->getMessage(),'ERR');
            $this->redis = $this->getRedis();
            sleep(1);
            return $this->keepFetchKey();
        }
    }

    function controlSubtask($cont){
        while(true){
            $this->assignTask($cont);

            $cont = $this->keepFetchKey();
            $this->show('sub Task begin:'.$cont);
            if(empty($cont)){
                $this->show('sub Task end');
                break;
            }
        }
        exit(0);
    }

    function assignTask($cont){
        $this->show("start assignTask:" . posix_getpid() ." $cont");
        $msgData = json_decode($cont,true);
        LogEX::i('更新推送信息群发data', ['data'=>$msgData]);
        $mateInfo = new WechatMaterial();
        //获取openid数据
        $setData = [
            'city'=>$msgData['city'],
            'page'=>$msgData['page'],
            'pageSize'=>$msgData['pageSize'],
        ];
        $ret = $mateInfo->getFansList($setData);
        $log['data'] = is_array($ret)?json_encode($ret):$ret;
        LogEX::i('更新推送信息群发openid', $log);
        $this->show("fetch openids:".json_encode($ret));
        if(empty($ret)){
            //失败
            $this->show('sendMessageFail: fetchOpenidFail:'.$cont);
            $this->keepFailKey($cont);
            return ;
        }
        //推送
        $data['openids'] = $ret;
        $data['timestamp'] = time();
        $data['Type'] = 'mpnews';
        $data['mediaId'] = $msgData['media_id'];
        $data['noticeId'] = $msgData['noticeId'];  //活动id
        LogEX::i('更新推送信息群发全部参数', ['data'=>$data]);
        $urlInfo = 'http://wx.push.mwee.cn/admin/weixin/sendmass';   //正式环境
        if (TP_DEBUG) $urlInfo = 'http://wx.push.9now.net/admin/weixin/sendmass';  //测试环境
        LogEX::i('更新推送信息群发结果信息', ['data'=>$urlInfo,'setData'=>json_encode($data)]);
        $rest = $mateInfo->post($urlInfo, json_encode($data));
        LogEX::i('微信推送发送信息', ['data' => $rest, 'url' => $urlInfo], __FILE__, __LINE__);
        $this->show("push result:".$rest);
        $result = json_decode($rest, true);
        $this->show('now push resutl:'.json_encode($result,JSON_UNESCAPED_UNICODE));
        if($result['statusCode'] == 200){
            $this->show('start insert mysql:'.$result['costTime']);
            $push = M('WechatPush');
            $dataInfo['media_id'] = $msgData['media_id'];
            $dataInfo['push_range'] = $msgData['push_range'];
            $dataInfo['msg_id'] = $result['data']['msg_id'];
            $pk = $push->add($dataInfo);
            $sql = $push->getLastSql();
            LogEX::i('微信推送指定标签下会员', ['sql' => $sql, 'dataInfo' => $dataInfo, 'url' => $this->url], __FILE__, __LINE__);

        }else{
            //失败
            $this->show('sendMessageFail: pushMessageError:'.$cont);
            $this->keepFailKey($cont);
        }
        $this->show("end assignTask:". posix_getpid() ." $cont");
    }

    function keepFailKey($data){
        $data = is_string($data)?$data:json_encode($data);
        try{
            $this->redis->lPush(self::FANS_INDEX_FAIL_KEY,$data);
        }catch (\Exception $e){
            $this->show('saveFailKeyFail:'.$e->getMessage().' data:'.$data);
        }
    }

    function show($msg,$level='DEBUG'){
//        $msg =  sprintf("[%s] %s\n",$level,$msg);
//        echo $msg;
        LogEX::i("MpushProcess",['data'=>$msg]);
    }

    function getRedis(){
        try{
            $redis = new \Redis;
            $redis->connect(REDIS_PUSH_TAGS,REDIS_PUSH_TAGS_PORT,5);
            $redis->auth(REDIS_PUSH_TAGS_AUTH);
            return $redis;
        }catch (\Exception $e){
            $this->show('connect redis error:'.$e->getMessage(),'ERR');
            return $this->getRedis();
        }
    }

    function err_handle(){
        if($this->ppid == posix_getpid()){
            $err = error_get_last();
            $this->show('error fatal :'. json_encode($err),'ERR');
            $this->redis = $this->getRedis();
            $this->init();
        }
    }

    function test(){
        foreach(range(0,1) as $i){
            $data['city'] = 258;
            $data['page'] = $i;
            $data['pageSize'] = 20;
            $data['noticeId'] = 12;  //活动id
            $data['media_id'] = 'Yv-b2NxsGyTYnN-XciEf7f_ffHBS_v8sfieLyhm6RtQ';
            $data['push_range'] = 258;
            $this->redis->lPush(self::FANS_INDEX_KEY,json_encode($data));
            $this->show('now in '.$i);
        }
    }

    function demo(){
        var_dump(C('DB_TYPE'));
        echo APP_STATUS;
    }
}